package com.example.mq.rocketmq.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * 拉取模式的消费者
 *
 * 在 RocketMQ 中，消费者可以使用两种不同的消费模式：推送模式（Push Mode）和拉取模式（Pull Mode）。这两种模式在消息消费的方式上有一些不同之处：
 *
 *     推送模式（Push Mode）：
 *         在推送模式下，消费者不断向消息服务器（Broker）发起拉取消息的请求，并通过长轮询（Long Polling）等机制来获取消息。
 *         消息服务器会根据消费者的请求情况主动推送消息给消费者。一旦有消息到达，消息服务器会立即将消息推送给消费者。
 *         推送模式适用于需要实时获取消息的场景，比如实时性要求较高的业务场景。
 *
 *     拉取模式（Pull Mode）：
 *         在拉取模式下，消费者定期向消息服务器发起拉取消息的请求，主动获取消息。
 *         消息服务器不会主动推送消息给消费者，而是等待消费者主动请求消息。
 *         拉取模式适用于消息消费者的处理能力较强、可以按需拉取消息的场景，比如批量处理消息或者需要预先处理消息的场景。
 *
 * 总的来说，推送模式和拉取模式的选择取决于你的业务需求和实际场景。推送模式适用于实时性要求较高的场景，可以更快地获取消息；而拉取模式适用于消息处理能力较强的场景，可以根据消费者的处理能力和需求来主动获取消息
 *
 * @author luowj12
 * @since 2021/5/14 14:44
 */
public class MQPullConsumer {

    //保存上一次消费的消息位置
    private static final Map offsetTable = new HashMap();
    public static void main(String[] args) throws MQClientException {

        //实例化pullConsumer
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("consumer_02");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.start();
        //拉取PullConsumerTopicTest topic下的所有消息队列
        Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("DEMO_01");

        //遍历消息队列
        for (MessageQueue mq : mqs) {
            System.err.println("Consume from the queue: " + mq);
            SINGLE_MQ:
            //
            while (true) {
                try {
                    //设置上次消费消息下标
                    PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
                    System.out.println(pullResult);
                    putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
                    switch (pullResult.getPullStatus()) {
                        //根据结果状态，如果找到消息，批量消费消息
                        case FOUND:
                            List<MessageExt> messageExtList = pullResult.getMsgFoundList();
                            for (MessageExt m : messageExtList) {
                                System.out.println(new String(m.getBody()));
                            }
                            break;
                        case NO_MATCHED_MSG:
                            break;
                        case NO_NEW_MSG:
                            break SINGLE_MQ;
                        case OFFSET_ILLEGAL:
                            break;
                        default:
                            break;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
        consumer.shutdown();
    }
    //保存上次消费的消息下标，这里使用了一个全局HashMap来保存
    private static void putMessageQueueOffset(MessageQueue mq, long offset) {
        offsetTable.put(mq, offset);
    }

    //获取上次消费的消息的下表
    private static long getMessageQueueOffset(MessageQueue mq) {
        Long offset = (Long) offsetTable.get(mq);
        if (offset != null) {
            return offset;
        }
        return 0;
    }

}
